package main

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"time"
)

type signal struct{}

type counter struct {
	sync.Mutex
	i int
}

type n_counter struct { // 基于无缓冲channel实现的锁机制
	c chan int
	i int
}

var cter counter

/**
 *	并行(parallelism), 指的就是在同一时刻，有两个或两个以上的任务(这里指进程)的代码在处理器上执行。
 *
 *  粗略看起来，多进程应用与单进程应用相比并没有什么质的提升。 设计成多进程应用犹豫将功能职责做了划分，并指定专门的模块来负责，所以从结构上来看，
 *  要比单进程更为清晰简洁，可读性与可维护性更好。
 * 这种将程序分成多个可独立执行的部分的结构化程序的设计方法，就是并发设计。采用了并发设计的应用也可以看成一组独立执行的模块的组合。
 **/

// 线程是运行与进程上下文中的更轻量级的执行流。
// 并发不是并行，并发关乎结构，并行关乎执行。

func spawn0(f func() error) <-chan error {
	c := make(chan error)

	go func() {
		c <- f()
	}()

	return c
}

func main() {
	c := spawn0(func() error {
		time.Sleep(2 * time.Second)
		return errors.New("timeout")
	})

	fmt.Println(<-c)

	ch1 := make(chan int)    // 无缓冲channel
	ch2 := make(chan int, 5) // 带缓冲channel

	// 发送与接收
	// channel 是用于Goroutine间通信的， 所以绝大多数对channel的读写是放在了不同的Goroutine中
	// ch1 <- 13  // 将整型字面值13发送到无缓冲channel类型变量ch1中
	go func() {
		ch1 <- 13 // 将发送操作放在另一个Goruntine执行
	}()
	n := <-ch1 // 从无缓冲channel类型变量ch1中接收一个整型值存储到整型变量n中, fatal error: all goroutines are asleep - deadlock!
	// 对同一个无缓冲 channel，只有对它进行接收操作的 Goroutine 和对它进行发送操作的 Goroutine 都存在的情况下，通信才能得以进行，否则单方面的操作会让对应的 Goroutine 陷入挂起状态
	// 对无缓冲 channel 类型的发送与接收操作，一定要放在两个不同的 Goroutine 中进行，否则会导致 deadlock。

	// 对带缓冲 channel 的发送操作在缓冲区未满、接收操作在缓冲区非空的情况下是异步的（发送或接收不需要阻塞等待）
	// 对一个带缓冲 channel 来说，在缓冲区未满的情况下，对它进行发送操作的 Goroutine 并不会阻塞挂起；
	// 在缓冲区有数据的情况下，对它进行接收操作的 Goroutine 也不会阻塞挂起。
	// 但当缓冲区满了的情况下，对它进行发送操作的 Goroutine 就会阻塞挂起；
	// 当缓冲区为空的情况下，对它进行接收操作的 Goroutine 也会阻塞挂起。
	ch2 <- 17  // 将整型字面值17发送到带缓冲channel类型变量ch2中
	m := <-ch2 // 从带缓冲channel类型变量ch2中接收一个整型值存储到整型变量m中

	println(n)
	println(m)

	// 使用操作符 <- 还可以声明只发送 channel类型(send-only)和只接收 channel 类型(recv-only)
	// c1 := make(chan<- int, 1) // 只发送channel
	// c2 := make(<-chan int, 1) // 只接收channel

	// <-c1     // invalid operation: <-c1 (receive from send-only type chan<- int)
	// c2 <- 13 // invalid operation: c2 <- 13 (send to receive-only type <-chan int)
	// 通常只发送channel和只接收channel是作为函数参数类型或返回值使用，方便限制外部操作

	ch := make(chan int, 5)
	var wg sync.WaitGroup
	wg.Add(2)

	go func() {
		produce(ch)
		wg.Done()
	}()

	go func() {
		consume(ch)
		wg.Done()
	}()

	wg.Wait()

	n = <-ch // 当channel被关闭, 接收到channel元素类型的零值
	println(n)
	m, ok := <-ch // 当channel被关闭, m接收到0值, ok为false
	println(m, ok)
	for v := range ch { // 当channel 被关闭, for range 循环直接结束
		println(v)
	}
	ch = make(chan int)

	// 当 select 语句中没有 default 分支，而且所有 case 中的 channel 操作都阻塞了的时候，整个 select 语句都将被阻塞，
	// 直到某一个 case 上的 channel 变成可发送，或者某个 case 上的 channel 变成可接收，select 语句才可以继续进行下去。
	select {
	case x := <-ch1:
		println("从channel ch1 中接收数据: ", x)
	case y, ok := <-ch2:
		fmt.Printf("从 channel ch2中接收数据: %d, 并根据ok值[%v]判断ch2是否已经关闭.\n", y, ok)
	case ch <- m:
		fmt.Printf("将m[%v]发送到channel ch中\n", m)
	default:
		println("当上面case中的channel通信都无法实现时, 执行该默认分支")
	}

	// 无缓冲channel的惯用法
	// 无缓冲 channel 兼具通信和同步特性，在并发程序中应用颇为广泛。
	// 第一种用法: 用作信号传递
	// 无缓冲 channel 用作信号传递的时候，有两种情况，分别是 1 对 1 通知信号和 1 对 n 通知信号。
	println("start a worker")
	c0 := spawn(worker)
	<-c0
	fmt.Println("worker work done")

	fmt.Println("start a group of workers")
	groupSignal := make(chan signal)
	c0 = spawnGroup(workerI, 5, groupSignal)
	time.Sleep(5 * time.Second)
	fmt.Println("the group of workers start to work...")
	close(groupSignal) // 关闭一个无缓冲 channel 会让所有阻塞在这个 channel 上的接收操作返回，从而实现了一种 1 对 n 的“广播”机制
	<-c0
	fmt.Println("the group of workers work Done!")

	// 第二种用法： 用于替代锁机制
	// 无缓冲 channel 具备同步特性，这让它在某些场合可以替代锁，让程序更加清晰， 可读性也更好
	// 传统锁机制

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			v := Increase()
			fmt.Printf("goroutine-%d: current counter value is %d\n", i, v)
			wg.Done()
		}(i)
	}

	wg.Wait()

	// 不要通过共享内存来通信，而是通过通信来共享内存
	c_ter := NewCounter()
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			v := c_ter.Increase()
			fmt.Printf("goroutine-%d: current counter value is %d\n", i, v)
			wg.Done()
		}(i)
	}

	wg.Wait()

	// 有缓冲区的惯用法
	// 带缓冲的 channel 与无缓冲的 channel 的最大不同之处，就在于它的异步性。
	// 对一个带缓冲 channel，在缓冲区未满的情况下，对它进行发送操作的 Goroutine 不会阻塞挂起；
	// 在缓冲区有数据的情况下，对它进行接收操作的 Goroutine 也不会阻塞挂起。
	// 第一种用法: 用于消息队列
	// channel 经常被 Go 初学者视为在多个 Goroutine 之间通信的消息队列，
	// 这是因为，channel 的原生特性与我们认知中的消息队列十分相似，包括 Goroutine 安全、有 FIFO（first-in, first out）保证等。

	// 无论是 1 收 1 发还是多收多发，带缓冲 channel 的收发性能都要好于无缓冲 channel；
	// 对于带缓冲 channel 而言，发送与接收的 Goroutine 数量越多，收发性能会有所下降；
	// 对于带缓冲 channel 而言，选择适当容量会在一定程度上提升收发性能。

	// 其实，和无缓冲 channel 更多用于信号 / 事件管道相比，可自行设置容量、异步收发的带缓冲 channel 更适合被用作为消息队列，
	// 并且，带缓冲 channel 在数据收发的性能上要明显好于无缓冲 channel
	// Go 支持 channel 的初衷是将它作为 Goroutine 间的通信手段，它并不是专门用于消息队列场景的。
	// 如果项目需要专业消息队列的功能特性，比如支持优先级、支持权重、支持离线持久化等，那么 channel 就不合适了，可以使用第三方的专业的消息队列实现。

	// 第二种用法: 用于计数信号量(counting sample)
	// Go 并发设计的一个惯用法，就是将带缓冲 channel 用作计数信号量（counting semaphore）。
	// 带缓冲 channel 中的当前数据个数代表的是，当前同时处于活动状态（处理业务）的 Goroutine 的数量，而带缓冲 channel 的容量（capacity），就代表了允许同时处于活动状态的 Goroutine 的最大数量。
	// 向带缓冲 channel 的一个发送操作表示获取一个信号量，而从 channel 的一个接收操作则表示释放一个信号量。

	active := make(chan struct{}, 3)
	jobs := make(chan int, 10)

	go func() {
		for i := 0; i < 8; i++ {
			jobs <- i
		}
		close(jobs)
	}()

	for j := range jobs {
		wg.Add(1)
		go func(j int) {
			active <- struct{}{}
			log.Printf("handle job: %d\n", j)
			time.Sleep(2 * time.Second)
			<-active
			wg.Done()
		}(j)
	}
	wg.Wait()

	// len 是Go语言的一个内置函数, 它支持数组、切片、map、字符串和channel类型的参数, 并返回对应的长度(一个整型值)
	// len(channel):
	// 当channel为无缓冲区channel时，总返回0
	// 当channel为有缓冲区channel时，返回当前channel未被读取的元素个数
	// 无法通过len函数判断channel满或空:
	// channel 原语用于多个 Goroutine 间的通信，一旦多个 Goroutine 共同对 channel 进行收发操作，
	// len(channel) 就会在多个 Goroutine 间形成“竞态”。
	// 单纯地依靠 len(channel) 来判断 channel 中元素状态，是不能保证在后续对 channel 的收发时 channel 状态是不变的。
	// 为了不阻塞在 channel 上，常见的方法是将“判空与读取”放在一个“事务”中，将“判满与写入”放在一个“事务”中，
	// 而这类“事务”我们可以通过 select 实现。
	wg.Add(2)
	go func(c chan<- int) {
		produce0(c)
		wg.Done()
	}(ch2)

	go func(c <-chan int) {
		consume0(c)
		wg.Done()
	}(ch2)

	wg.Wait()

	// 在不改变 channel 状态的前提下，单纯地侦测 channel 的状态，而又不会因 channel 满或空阻塞在 channel 上。
	// 但很遗憾，目前没有一种方法可以在实现这样的功能的同时，适用于所有场合。
	// 有多个发送者，但有且只有一个接收者, 可以在接收端通过len函数判断channel状态
	// 有多个接收者，但有且只有一个发送者, 可以通过len函数和cap函数的值判断channel是否已满, 再进行是否发送数据的判断
}

func Increase() int {
	cter.Lock()
	defer cter.Unlock()
	cter.i++
	return cter.i
}

/**
 *	生产函数
 **/
func produce(ch chan<- int) {
	for i := 0; i < 10; i++ {
		ch <- i + 1
		time.Sleep(time.Second)
	}
	close(ch) // channel 使用惯例: 发送端关闭channel
	// 发送端没有像接收端那样可以安全判断channel是否关闭的方法
	// 发送端继续发送时会panic
}

/**
 *	消费函数
 **/
func consume(ch <-chan int) {
	for n := range ch {
		fmt.Printf("消费到数据: %d\n", n)
	}
}

func worker() {
	println("worker is working...")
	time.Sleep(time.Second)
}

func spawn(f func()) <-chan signal {
	c := make(chan signal) // 无缓冲channel
	go func() {
		println("worker start to work...")
		f()
		c <- signal{}
	}()
	return c
}

func workerI(i int) {
	fmt.Printf("worker %d: is working...\n", i)
	time.Sleep(time.Second)
	fmt.Printf("worker %d work done...\n", i)
}

func spawnGroup(f func(i int), num int, groupSignal <-chan signal) <-chan signal {
	c := make(chan signal)
	var wg sync.WaitGroup

	for i := 0; i < num; i++ {
		wg.Add(1)
		go func(i int) {
			<-groupSignal
			fmt.Printf("worker %d start to work...\n", i)
			f(i)
			wg.Done()
		}(i + 1)
	}

	go func() {
		wg.Wait()
		c <- signal{}
	}()

	return c
}

func NewCounter() *n_counter {
	cter := &n_counter{
		c: make(chan int),
	}
	go func() {
		for {
			cter.i++
			cter.c <- cter.i
		}
	}()
	return cter
}

func (cter *n_counter) Increase() int {
	return <-cter.c
}

func tryRecv(c <-chan int) (int, bool) {
	select {
	case i := <-c:
		return i, true
	default:
		return 0, false
	}
}

func trySend(c chan<- int, i int) bool {
	select {
	case c <- i:
		return true
	default:
		return false
	}
}

func produce0(c chan<- int) {
	var i int = 1
	for i < 20 {
		time.Sleep(2 * time.Second)
		ok := trySend(c, i)
		if ok {
			fmt.Printf("[producer]: send [%d] to channel.\n", i)
			i++
		} else {
			fmt.Printf("[producer]: try send [%d], but channel is full.\n", i)
		}
	}
}

func consume0(c <-chan int) {
	for {
		i, ok := tryRecv(c)
		if ok {
			fmt.Printf("[consumer]: recv [%d] from channel.\n", i)
			if i >= 10 {
				fmt.Println("[consumer]: exit")
				return
			}
		} else {
			fmt.Println("[consumer]: try to recv from channel, but the channel is empty")
			time.Sleep(time.Second)
		}
	}
}
